package com.wei.web.middleware.retry.sequence;

import com.alibaba.fastjson.JSONObject;
import com.github.pagehelper.ISelect;
import com.github.pagehelper.Page;
import com.github.pagehelper.PageHelper;
import com.google.common.base.Joiner;
import com.tuya.basic.mq.IMqKafkaProducer;
import com.tuya.basic.mq.domain.KafkaMqData;
import com.tuya.bay.bee.common.enums.KafkaMessageStatusEnum;
import com.tuya.bay.bee.core.constants.RedisConstants;
import com.tuya.bay.bee.core.generated.dao.BeeSendKafkaMessageMapper;
import com.tuya.bay.bee.core.generated.dataobject.BeeSendKafkaMessageDO;
import com.tuya.bay.bee.core.seqqueue.SequenceQueueProcessor;
import com.tuya.bay.bee.core.threads.AsyncTaskManager;
import com.tuya.bay.bee.utils.LockUtil;
import com.tuya.loki.client.LokiClient;
import com.tuya.middleware.dubai.annotation.Master;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.concurrent.ListenableFutureCallback;

import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.TimeUnit;


/**
 * @author pangqun
 * @date 2021-06-11
 * @description kafka任务队列处理器
 */
@Slf4j
@AllArgsConstructor
@Component
public class KafkaSequenceQueueProcessor implements SequenceQueueProcessor<BeeSendKafkaMessageDO> {
    private final BeeSendKafkaMessageMapper beeSendKafkaMessageMapper;
    private final IMqKafkaProducer<JSONObject> mqKafkaProducer;
    private final AsyncTaskManager asyncTaskManager;
    private final LokiClient lokiClient;

    public static final String REDIS_TRIGGER_AUTOMATIC_PROCESS_KEY = "trigger_automatic_process";
    public static final String REDIS_EXECUTE_AUTOMATIC_PROCESS_KEY = "execute_automatic_process";

    private static final Long REDIS_EXECUTE_AUTOMATIC_PROCESS_KEY_LOCK_WAIT_TIME = 10000L;
    private static final Long REDIS_EXECUTE_AUTOMATIC_PROCESS_KEY_TIMEOUT = 10000L;

    private final long scanTime = 300000L;

    public void put(String msgTopic, String topicKey, String msgContent) {
        log.debug("put msgTopic {}, topicKey {}, msgContent {}",
                msgTopic, topicKey, msgContent);
        BeeSendKafkaMessageDO data = new BeeSendKafkaMessageDO();
        data.setMsgTopic(msgTopic);
        data.setTopicKey(topicKey);
        data.setMsgContent(msgContent);
        put(data);
    }


    @Override
    @Transactional(rollbackFor = Exception.class)
    public void put(BeeSendKafkaMessageDO data) {
        try {
            beeSendKafkaMessageMapper.insert(data);
        } catch (Exception e) {
            log.error("beeSendKafkaMessageMapper.insert failed", e);
            throw new RuntimeException(e);
        }


    }

    /**
     * 尝试触发发送消息的任务。
     * 获取第一条没有发送成功的消息，如果是未发送，则尝试设置状态。如果设置成功，则开始触发任务发送。
     * 否则说明已经有其他线程处理，直接退出。
     *
     * @param topicKey
     * @param msgTopic
     */
    @Override
    @Master
    public void tryAutomaticProcess(String topicKey, String msgTopic) {
        String topicAndKeyPair = Joiner.on("_").join("topic", msgTopic, "msg_key", topicKey);

        /**
         * 触发事件key
         */
        String triggerLockKey = Joiner.on("_").join(RedisConstants.REDIS_PREFIX,
                REDIS_TRIGGER_AUTOMATIC_PROCESS_KEY,
                topicAndKeyPair);

        /**
         * 如果setnx设置成功，说明需要进入触发执行
         * 如果setnx设置失败，说明已经有触发事件，准备执行，自己不需要设置了，直接退出
         */
        Boolean setTriggerResult = lokiClient.opsForValue().setIfAbsent(triggerLockKey, "1", 10, TimeUnit.SECONDS);
        log.info("tryAutomaticProcess topicKey:{} msgTopic:{} redisKey:{} triggerLockKey:{} setTriggerResult:{}", topicKey, msgTopic,
                topicAndKeyPair,
                triggerLockKey, setTriggerResult);
        if (setTriggerResult != null && setTriggerResult) {
            /**
             * 开始执行任务
             * 执行锁，只能有一个线程执行
             */
            String executeLockKey =
                    Joiner.on("_").join(RedisConstants.REDIS_PREFIX,
                            REDIS_EXECUTE_AUTOMATIC_PROCESS_KEY, topicAndKeyPair);

            LockUtil.executeWithWaitTime(executeLockKey, () -> {
                        lokiClient.delete(triggerLockKey);

                        BeeSendKafkaMessageDO firstUnsendMessage = tryGetFirstUnsendMessage(topicKey, msgTopic);
                        if (firstUnsendMessage != null) {
                            asyncTaskManager.execute(() -> {
                                doAutomaticSerialSendMessage(topicKey, msgTopic, firstUnsendMessage);
                            });
                        }
                        return null;
                    }, REDIS_EXECUTE_AUTOMATIC_PROCESS_KEY_LOCK_WAIT_TIME,
                    REDIS_EXECUTE_AUTOMATIC_PROCESS_KEY_TIMEOUT);
        }

    }

    /**
     * 获取第一条没有发送成功的消息，如果是未发送，则尝试设置状态。如果设置成功，则开始触发任务发送。
     *
     * @param topicKey
     * @param msgTopic
     * @return
     */
    public BeeSendKafkaMessageDO tryGetFirstUnsendMessage(String topicKey, String msgTopic) {
        log.info(" get first record topicKey:{} msgTopic:{}", topicKey,
                msgTopic);
        BeeSendKafkaMessageDO beeSendKafkaMessageDO = beeSendKafkaMessageMapper.selectByTopicKeyAndMsgTopic(topicKey, msgTopic);
        if (beeSendKafkaMessageDO == null) {
            log.debug("tryAutomaticProcess lock to get null topicKey:{} msgTopic:{}", topicKey,
                    msgTopic);
            return null;
        }

        log.debug("get UNSENT topicKey:{} msgTopic:{} id:{}", topicKey,
                msgTopic, beeSendKafkaMessageDO.getId());
        if (!beeSendKafkaMessageDO.getStatus().equals(KafkaMessageStatusEnum.UNSENT.getValue())) {
            return null;
        }
        beeSendKafkaMessageDO.setStatus(KafkaMessageStatusEnum.SENT_ING.getValue());
        int ret = beeSendKafkaMessageMapper.updateSending(beeSendKafkaMessageDO);
        if (ret <= 0) {
            return null;
        }
        return beeSendKafkaMessageDO;
    }

    /**
     * 获取当前发送消息的下一条消息，如果是未发送，则尝试设置状态。如果设置成功，则继续发送。
     *
     * @param topicKey
     * @param msgTopic
     * @param kafkaMessageDO
     * @return
     */
    public BeeSendKafkaMessageDO tryGetNextMessageForUnsend(String topicKey, String msgTopic, BeeSendKafkaMessageDO kafkaMessageDO) {
        log.info("tryAutomaticProcess lock to get first record topicKey:{} msgTopic:{}", topicKey,
                msgTopic);
        BeeSendKafkaMessageDO nextKafkaMessageDO =
                beeSendKafkaMessageMapper.selectNextMessage(topicKey, msgTopic, kafkaMessageDO.getId());
        if (nextKafkaMessageDO == null) {
            log.debug("tryAutomaticProcess lock to get null topicKey:{} msgTopic:{}", topicKey,
                    msgTopic);
            return null;
        }

        log.debug("tryAutomaticProcess lock to get SENT_ING topicKey:{} msgTopic:{} id:{}", topicKey,
                msgTopic, nextKafkaMessageDO.getId());
        if (!nextKafkaMessageDO.getStatus().equals(KafkaMessageStatusEnum.UNSENT.getValue())) {
            return null;
        }
        nextKafkaMessageDO.setStatus(KafkaMessageStatusEnum.SENT_ING.getValue());
        int ret = beeSendKafkaMessageMapper.updateSending(nextKafkaMessageDO);
        if (ret <= 0) {
            return null;
        }
        return nextKafkaMessageDO;
    }

    /**
     * 顺序的发送当前topic key下，未发送的消息。
     * 这个方法必须是顺序发送当前topic key下的消息。
     * 如果尝试设置消息状态失败，则表示有其他线程在发送当前topic key，防止有多线程发送，则直接退出。
     *
     * @param topicKey
     * @param msgTopic
     * @param kafkaMessageDO
     */
    public void doAutomaticSerialSendMessage(String topicKey, String msgTopic, BeeSendKafkaMessageDO kafkaMessageDO) {
        BeeSendKafkaMessageDO currentMessageDO = kafkaMessageDO;
        BeeSendKafkaMessageDO lastSendMessageDO = null;
        List<SendMessageCallbackContext> sendMessageCallbackContextList = new LinkedList<>();

        do {
            SendMessageCallbackContext sendMessageCallbackContext = doAutomaticSendMessage(topicKey, msgTopic, currentMessageDO);
            if (sendMessageCallbackContext.getSendResult() == SendMessageResult.SEND_FAILED) {
                break;
            } else {
                sendMessageCallbackContextList.add(sendMessageCallbackContext);
            }
            lastSendMessageDO = currentMessageDO;

            // 循环等待，拿下一条待发送消息
            CheckIfSendResult getResult = CheckIfSendResult.STOP;
            while (true) {
                CheckAllMessageSendResult checkAllMessageSendResult = checkAllSendingMessageResult(sendMessageCallbackContextList);
                if (checkAllMessageSendResult.isHasMessageSendFailed()) {
                    currentMessageDO = null;
                    getResult = CheckIfSendResult.STOP;
                    break;
                }

                currentMessageDO = tryGetNextMessageForUnsend(topicKey, msgTopic, lastSendMessageDO);
                if (currentMessageDO != null) {
                    getResult = CheckIfSendResult.GOON;
                    break;
                }

                if (checkAllMessageSendResult.isAllMessageSendComplete()) {
                    currentMessageDO = tryGetNextMessageForUnsend(topicKey, msgTopic, lastSendMessageDO);
                    if (currentMessageDO != null) {
                        getResult = CheckIfSendResult.GOON;
                        break;
                    } else {
                        currentMessageDO = null;
                        getResult = CheckIfSendResult.STOP;
                        break;
                    }
                }

                try {
                    Thread.sleep(10);
                } catch (InterruptedException e) {
                    log.error("sleep failed", e);
                }
            }


            if (CheckIfSendResult.GOON != getResult) {
                break;
            }
        } while (true);

        return;
    }


    /**
     * 检查已经发送的消息的发送结果，是否全部发送完成，是否有发送失败的消息
     *
     * @param sendMessageCallbackContextList
     * @return
     */
    public CheckAllMessageSendResult checkAllSendingMessageResult(List<SendMessageCallbackContext> sendMessageCallbackContextList) {
        CheckAllMessageSendResult allMessageSendResult = new CheckAllMessageSendResult();
        allMessageSendResult.setAllMessageSendComplete(true);
        allMessageSendResult.setHasMessageSendFailed(false);

        Iterator<SendMessageCallbackContext> it = sendMessageCallbackContextList.iterator();
        while (it.hasNext()) {
            SendMessageCallbackContext sendMessageCallbackContext = it.next();
            if (sendMessageCallbackContext.getSendResult() == SendMessageResult.SEND_FAILED) {
                allMessageSendResult.setHasMessageSendFailed(true);
            }

            if (sendMessageCallbackContext.getSendResult() == SendMessageResult.SENDING) {
                allMessageSendResult.setAllMessageSendComplete(false);
            }
        }
        return allMessageSendResult;
    }

    /**
     * 异步发送消息
     *
     * @param topicKey
     * @param msgTopic
     * @param kafkaMessageDO
     * @return
     */
    private SendMessageCallbackContext doAutomaticSendMessage(String topicKey, String msgTopic, BeeSendKafkaMessageDO kafkaMessageDO) {
        SendMessageCallbackContext sendMessageCallbackContext = new SendMessageCallbackContext();
        sendMessageCallbackContext.setMessageId(kafkaMessageDO.getId());
        sendMessageCallbackContext.setSendResult(SendMessageResult.SENDING);

        KafkaMqData<JSONObject> message = new KafkaMqData<>(JSONObject.parseObject(kafkaMessageDO.getMsgContent()));

        ListenableFutureCallback<SendResult<String, KafkaMqData<JSONObject>>> listenableFutureCallback =
                new ListenableFutureCallback<SendResult<String, KafkaMqData<JSONObject>>>() {
                    @Override
                    public void onSuccess(SendResult<String, KafkaMqData<JSONObject>> stringKafkaMqDataSendResult) {
                        log.info("onSuccess topicKey:{} msgTopic:{} Id:{}", topicKey,
                                msgTopic, kafkaMessageDO.getId());
                        kafkaMessageDO.setStatus(KafkaMessageStatusEnum.SENT.getValue());
                        beeSendKafkaMessageMapper.updateSendStatusAndTryNum(kafkaMessageDO);
                        sendMessageCallbackContext.setSendResult(SendMessageResult.SEND_SUCCESS);
                    }

                    @Override
                    public void onFailure(Throwable throwable) {
                        log.error("onFailure topicKey:{} msgTopic:{} Id:{}", topicKey,
                                msgTopic, kafkaMessageDO.getId());
                        log.error("onFailure", throwable);
                        kafkaMessageDO.setStatus(KafkaMessageStatusEnum.FAIL_SENT.getValue());
                        beeSendKafkaMessageMapper.updateSendStatusAndTryNum(kafkaMessageDO);
                        sendMessageCallbackContext.setSendResult(SendMessageResult.SEND_FAILED);
                    }
                };

        try {
            log.info("tryAutomaticProcess asyncSend topicKey:{} msgTopic:{} Id:{}", topicKey,
                    msgTopic, kafkaMessageDO.getId());
            mqKafkaProducer.asyncSend(kafkaMessageDO.getMsgTopic(), message, topicKey, listenableFutureCallback);
        } catch (Exception e) {
            log.error("mqKafkaProducer.asyncSend failed: " + kafkaMessageDO.getMsgTopic(), e);
            kafkaMessageDO.setStatus(KafkaMessageStatusEnum.FAIL_SENT.getValue());
            beeSendKafkaMessageMapper.updateSendStatusAndTryNum(kafkaMessageDO);
            sendMessageCallbackContext.setSendResult(SendMessageResult.SEND_FAILED);
        }
        return sendMessageCallbackContext;
    }


    @Override
    @Master
    public void processResultCallback(String topicKey, String msgTopic, String requestId) {
        throw new RuntimeException("no supported!");
    }

    @Override
    @Master
    public void onTimerScan() {
        log.info("start timer");
        recoverySend();
    }

    @Master
    public void recoverySend() {
        long time = System.currentTimeMillis() - scanTime;
        ISelect iSelect = () -> beeSendKafkaMessageMapper.selectMessageForRetrySend(time);
        Page<BeeSendKafkaMessageDO> page = PageHelper.startPage(1, 100)
                .doSelectPage(iSelect);

        for (int i = 1; i <= page.getPages(); i++) {
            Page<BeeSendKafkaMessageDO> sendKafkaMessageDOS = PageHelper.startPage(i, 100)
                    .doSelectPage(iSelect);

            for (BeeSendKafkaMessageDO beeSendKafkaMessageDO : sendKafkaMessageDOS.getResult()) {
                log.info("recovery send for TopicKey:{} MsgTopic:{}",
                        beeSendKafkaMessageDO.getTopicKey(), beeSendKafkaMessageDO.getMsgTopic());
                BeeSendKafkaMessageDO firstFailedMessageDO =
                        beeSendKafkaMessageMapper.selectByTopicKeyAndMsgTopic(
                                beeSendKafkaMessageDO.getTopicKey(), beeSendKafkaMessageDO.getMsgTopic());

                int updateNum = beeSendKafkaMessageMapper.
                        updateStatusForRetry(beeSendKafkaMessageDO.getTopicKey(),
                                beeSendKafkaMessageDO.getMsgTopic(),
                                firstFailedMessageDO.getId(), System.currentTimeMillis());
                log.info("recovery send to update status for TopicKey:{} MsgTopic:{} msgId:{} updateNum:{}",
                        beeSendKafkaMessageDO.getTopicKey(), beeSendKafkaMessageDO.getMsgTopic(),
                        firstFailedMessageDO.getId(), updateNum);

                tryAutomaticProcess(beeSendKafkaMessageDO.getTopicKey(),
                        beeSendKafkaMessageDO.getMsgTopic());

            }
        }

    }

    public Integer deleteOldSuccessMessages(Long gmtModified, Integer pageSize) {
        return beeSendKafkaMessageMapper.deleteOldSuccessMessages(gmtModified, pageSize);
    }


    public enum CheckIfSendResult {
        STOP,
        GOON
    }

    @Data
    public static class SendMessageCallbackContext {
        private Long messageId;
        private volatile SendMessageResult sendResult;
    }

    public enum SendMessageResult {
        SENDING,
        SEND_SUCCESS,
        SEND_FAILED
    }

    @Data
    public static class CheckAllMessageSendResult {
        private boolean isAllMessageSendComplete;
        private boolean hasMessageSendFailed;
    }

}
